#ifndef __REDUCE_H__
#define __REDUCE_H__

#include <cstdlib>
#include <iostream>
#include <pthread.h>
#include <assert.h>
#include <cstdio>
#include "global.h"

using namespace std;

template <typename D, typename T>
class Reducer {
public:
  T nodeval[99];
  D* data;
  INDEX length;

  T result;

  int threadid[99];
  pthread_t threads[99];

  Reducer(D*, INDEX);
  void Start();
  void* Reduce(void*);

  virtual void init(T&) = 0;
  virtual void accum(T& t, D d) = 0;
  virtual void combine(T& left, T right) = 0;
  virtual T reduce(T t) = 0;

  static void* ReduceWrapper(void* pt_pair) {
    return ((Reducer<D, T>*) ((PtPair*)pt_pair)->This)->Reduce(((PtPair*)pt_pair)->threadid);
    // Pthread requires static function without "this" pointer
    // But functions called by Reduce needs this
    // Using method mentioned here to get around:
    // http://stackoverflow.com/questions/1151582/pthread-function-from-a-class
    // Answered by Jeremy Friesner
  }
};

template <typename D, typename T>
Reducer<D, T>::Reducer(D* _data, INDEX _length) {
  data = _data;
  length = _length;
  //cout<<_length<<" "<<P<<endl;
  fflush(stdout);
  assert(_length % P == 0);

  for(int i=0; i<P; i++)
    threadid[i] = i;

  pthread_mutex_init(&lock, NULL);
  pthread_barrier_init(&barrier, NULL, P);
}

template <typename D, typename T>
void Reducer<D, T>::Start() {
  int rc;
  PtPair pt_pair[99];

  for(int i=0; i<P; i++) {
    pt_pair[i].This = this;
    pt_pair[i].threadid = (void*)(threadid+i);

    rc = pthread_create(&threads[i], NULL, ReduceWrapper, (void*)(&pt_pair[i]));
    if (rc)
      exit(-1);
  }
  for(int i=0; i<P; i++) {
    rc = pthread_join(threads[i], NULL);
  }
  return;
}

template <typename D, typename T>
void* Reducer<D, T>::Reduce(void* threadid_pt) {
  int threadid = *(int*)threadid_pt;
  INDEX base = length/P * threadid;
  INDEX size = length/P;
  T tally;
  int stride = 1;

  // local accumulation
  init(tally);
  init(nodeval[threadid]);

  for(INDEX i=0; i<size; i++)
    accum(tally, data[base+i]);
  nodeval[threadid] = tally;

  // moving up in the tree, converging results
  while(stride < P) {
    pthread_barrier_wait(&barrier);
    if (threadid%(2*stride) == 0) {
      combine(nodeval[threadid], nodeval[threadid+stride]);
    }
    stride = 2*stride;
  }

  // converge to a single number
  if (threadid == 0) {
    init(result);
    result = reduce(nodeval[0]);
  }

  //pthread_mutex_lock (&lock);
  //cout<<"finishing thread: "<<threadid<<endl;
  //pthread_mutex_unlock (&lock);

  return NULL;
}

#endif

